package net.bw.realtime.tmall.dws.job;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;

/*
 * @ Author：liuyawei
 * @ Date：2025-05-26
 */
public class TmallLogProvinceHourWindowDwsJob {

    public static void main(String[] args) {

        // 1. 创建执行环境
        TableEnvironment tabEnv = getTableEnv();

        // 2. 创建输入表,映射到kafka消息队列
        createInputTable(tabEnv);

        // 3. 处理数据
        handle(tabEnv);

        // 4.创建输出表,映射到clickhouse
        createOutputTable(tabEnv);

        // 5.保存数据
        saveToClickHouse(tabEnv);

    }

    private static void handle(TableEnvironment tabEnv) {

        Table province_table = tabEnv.sqlQuery(
                "SELECT " +
                        " `location`['province'] AS province," +
                        " `location`['city'] AS city," +
                        " CAST(`actionDetail`['price'] AS BIGINT) AS price," +
                        " row_time" +
                        " FROM dwd_tmall_log_kafka_source"
        );

        //province_table.execute().print();

        tabEnv.createTemporaryView("province_table", province_table);

        Table province_report_table = tabEnv.sqlQuery(
                "SELECT\n" +
                        "  TUMBLE_START(row_time, INTERVAL '1' HOURS) AS window_start_time,\n" +
                        "  TUMBLE_END(row_time, INTERVAL '1' HOURS) AS window_end_time,\n" +
                        "  province,\n" +
                        "  sum(price) AS `total_amount`,\n" +
                        "  UNIX_TIMESTAMP() * 1000 AS ts\n" +
                        "  FROM province_table\n" +
                        "GROUP BY\n" +
                        "  TUMBLE(row_time, INTERVAL '1' HOURS)," +
                        "  province\n"
        );

        //province_report_table.execute().print();

        tabEnv.createTemporaryView("province_report_table", province_report_table);


        Table city_report_table = tabEnv.sqlQuery(
                "SELECT\n" +
                        "  TUMBLE_START(row_time, INTERVAL '1' HOURS) AS window_start_time,\n" +
                        "  TUMBLE_END(row_time, INTERVAL '1' HOURS) AS window_end_time,\n" +
                        "  city,\n" +
                        "  sum(price) AS `total_amount`,\n" +
                        "  UNIX_TIMESTAMP() * 1000 AS ts\n" +
                        "  FROM province_table\n" +
                        "GROUP BY\n" +
                        "  TUMBLE(row_time, INTERVAL '1' HOURS)," +
                        "  city\n"
        );

        //city_report_table.execute().print();

        tabEnv.createTemporaryView("city_report_table", city_report_table);

    }

    private static void saveToClickHouse(TableEnvironment tabEnv) {

        tabEnv.executeSql(
                "INSERT INTO dws_province_daily_report_clickhouse_sink\n" +
                        "SELECT\n" +
                        "    DATE_FORMAT(window_start_time, 'yyyy-MM-dd HH:mm:ss') AS window_start_time\n" +
                        "    , DATE_FORMAT(window_end_time, 'yyyy-MM-dd HH:mm:ss') AS window_end_time\n" +
                        "    , `province`\n" +
                        "    , `total_amount`\n" +
                        "    , ts\n" +
                        "FROM province_report_table"
        );


        tabEnv.executeSql(
                "INSERT INTO dws_city_daily_report_clickhouse_sink\n" +
                        "SELECT\n" +
                        "    DATE_FORMAT(window_start_time, 'yyyy-MM-dd HH:mm:ss') AS window_start_time\n" +
                        "    , DATE_FORMAT(window_end_time, 'yyyy-MM-dd HH:mm:ss') AS window_end_time\n" +
                        "    , `city`\n" +
                        "    , `total_amount`\n" +
                        "    , ts\n" +
                        "FROM city_report_table"
        );

    }

    private static void createOutputTable(TableEnvironment tabEnv) {

        tabEnv.executeSql(
                "CREATE TABLE dws_province_daily_report_clickhouse_sink (\n" +
                        "  `window_start_time` STRING COMMENT '窗口开始日期时间',\n" +
                        "  `window_end_time` STRING COMMENT '窗口结束日期时间',\n" +
                        "  `province` STRING COMMENT '省份',\n" +
                        "  `total_amount` BIGINT COMMENT '成交额',\n" +
                        "  `ts` BIGINT COMMENT '时间戳'\n" +
                        ") WITH (\n" +
                        "  'connector' = 'clickhouse',\n" +
                        "  'url' = 'jdbc:clickhouse://node103:8123/jtp_tmall_report',\n" +
                        "  'table' = 'dws_province_daily_report',\n" +
                        "  'username' = 'default',\n" +
                        "  'password' = '',\n" +
                        "  'format' = 'json'\n" +
                        ")"
        );

        tabEnv.executeSql(
                "CREATE TABLE dws_city_daily_report_clickhouse_sink (\n" +
                        "  `window_start_time` STRING COMMENT '窗口开始日期时间',\n" +
                        "  `window_end_time` STRING COMMENT '窗口结束日期时间',\n" +
                        "  `city` STRING COMMENT '城市',\n" +
                        "  `total_amount` BIGINT COMMENT '成交额',\n" +
                        "  `ts` BIGINT COMMENT '时间戳'\n" +
                        ") WITH (\n" +
                        "  'connector' = 'clickhouse',\n" +
                        "  'url' = 'jdbc:clickhouse://node103:8123/jtp_tmall_report',\n" +
                        "  'table' = 'dws_city_daily_report',\n" +
                        "  'username' = 'default',\n" +
                        "  'password' = '',\n" +
                        "  'format' = 'json'\n" +
                        ")"
        );

    }

    private static void createInputTable(TableEnvironment tabEnv) {
        // 映射到kafka消息队列
        tabEnv.executeSql("CREATE TABLE dwd_tmall_log_kafka_source\n" +
                "(\n" +
                "    orderId         STRING COMMENT '订单ID',\n" +
                "    userId          STRING COMMENT '用户ID',\n" +
                "    actionType      STRING COMMENT '行为类型',\n" +
                "    `actionDetail`  MAP<STRING, STRING> COMMENT '商品信息',\n" +
                "    `deviceInfo`    MAP<STRING, STRING> COMMENT '设备信息',\n" +
                "    `location`      MAP<STRING, STRING> COMMENT '地理信息',\n" +
                "    `session`       MAP<STRING, STRING> COMMENT '会话信息',\n" +
                "    `timestamp`     BIGINT,\n" +
                "    row_time AS TO_TIMESTAMP_LTZ(`timestamp`, 3),\n" +
                "    WATERMARK FOR row_time AS row_time - INTERVAL '0' HOUR\n" +
                ") WITH (\n" +
                "    'connector' = 'kafka',\n" +
                "    'topic' = 'dwd-tmall-log',\n" +
                "    'properties.bootstrap.servers' = 'node101:9092,node102:9092,node103:9092',\n" +
                "    'properties.group.id' = 'gid_dws_tmall_log',\n" +
                "    'scan.startup.mode' = 'earliest-offset',\n" +
                "    'format' = 'json',\n" +
                "    'json.fail-on-missing-field' = 'false',\n" +
                "    'json.ignore-parse-errors' = 'true'\n" +
                ")"
        );
    }

    private static TableEnvironment getTableEnv() {

        //  1. 创建表执行环境
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .useBlinkPlanner()
                .inStreamingMode()
                .build();

        TableEnvironment tabEnv = TableEnvironment.create(settings);

        // 2. 配置属性
        Configuration configuration = tabEnv.getConfig().getConfiguration();
        configuration.setString("table.local-time-zone", "Asia/Shanghai");
        configuration.setString("table.exec.resource.default-parallelism", "1");
        configuration.setString("table.exec.state.ttl", "5 s");

        return tabEnv;

    }


}
